/*!
后台服务接收文件任务，执行：文本提取，创建索引，文件同步等。

1. post MQ for m/s sync
2. if slave:
    2.1 timer to check stat for m/s sync
    2.2 read mq for m/s sync
3. extract file as fidx.rabbit.txt and build index

*/
use crate::*;
use rabbitmq::client::ChannelClient;
use sha2::{Digest, Sha256};
use std::fs::OpenOptions;
use std::io::Write;
use tokio::sync::mpsc;

use tokio::time::Duration;
//use bytes::{Buf, Bytes};
use log::Level::{Error,Info, Debug};
#[cfg(feature = "rotatelog")]
use log::{debug, error, log as event};
#[cfg(feature = "tracelog")]
use tracing::{debug, error, event };

/**
axum handle中接到‘重’任务后，通过channel通知worker，发送对象就是Task
 */
#[allow(dead_code)]
pub enum Task {
    /**
    新写入文件
     */
    New(GfsResponse),
    /**
    写入记录文件
     */
    Record(RecordParam, String, bytes::Bytes),
    /**
    新同步指令
     */
    Sync(GfsResponse),
}

impl Task {
    /**
    接收到的指令，执行相应的任务
     * 这里通过 match ... ref mut 进行了‘可变对象‘解构
     * 需要提醒的是，run传递的参数不能是run(&self...)，而应该 run(&mut self...)
     */
    pub async fn run(
        &mut self,
        shared: &ShareComm,
        mq_actor: &MqActor,        
    ) -> Result<()> {
        match self {
            Task::New(ref mut todo) => {
                if shared.is_master() {
                    // current is master, send sync MQ
                    let ids = format!("{}", todo.idx);
                    let val = serde_json::to_string(&todo)?;
                    mq_actor.say_sync(TOPIC, &ids, &val).await?;
                }
                extract_and_index(shared, &todo).await?;
            }

            Task::Record(ref mut todo, command, body) => {
                // event!(target: "parallel", Level::ERROR, "command = {}, {:#?}", command,  todo );
                let filename =
                    gfs_hashcode_to_path(&shared.gfs.root, todo.ihash, todo.idx).join(&todo.name);

                let task_option = if command.as_str().eq(RECORD_CMD_ERROR) {
                    TASK_STATUS_ERRO
                } else if command.as_str().eq(RECORD_CMD_NONE) {
                    TASK_STATUS_NONE
                } else if command.as_str().eq(RECORD_CMD_DONE) {
                    TASK_STATUS_DONE
                } else if body.len() > 0 {
                    TASK_STATUS_DONE
                } else {
                    TASK_STATUS_NONE
                };
                /*
                实际存在异步不同线程操作相同文件的可能，所以借助record，防止文件操作冲突
                但也大大降低了整体的效率
                考虑到record不是本系统的核心，应用场景不需要高并发，故此方案较单有效。

                 */
                let mut record = shared.record.write().await;

                if todo.ikey >= 0 {
                    event!(target: "parallel", loglevel(Debug), "ikey = {}, status = {:#?}", todo.ikey, task_option);
                    let ikey = todo.ikey as usize;
                    record.set_status(&todo.taskname, task_option, ikey);
                }
                if (command.as_str().eq(RECORD_CMD_CLEAR)
                    || command.as_str().eq(RECORD_CMD_OVERD)) 
                    && todo.ikey >=0 { 
                    error!(
                        "remove cache {}, create new {:#?}",
                        &todo.taskname, &filename
                    );
                    record.remove(&todo.taskname);
                }

                event!(target: "parallel", loglevel(Debug), "record file = {:#?}, ikey = {}, command={}", filename, todo.ikey, command);
                if command.as_str().eq(RECORD_CMD_CLEAR)
                    || command.as_str().eq(RECORD_CMD_OVERD)
                    || !filename.exists()
                {
                    let mut file = std::fs::File::create(filename)?;
                    file.write_all(body)?;
                } else {
                    // filename.exists() and append
                    if body.len() > 0 {
                        let mut file = OpenOptions::new().append(true).open(filename)?;
                        file.write_all(body)?;
                    }
                }

                /*  record不同于文件，粒度较小，所以不需要频繁的同步，系统提供了单独的同步指令，应用者可自行调用！
                //
                if shared.is_master() {
                    ...
                    mq_actor.say_sync(TOPIC, &ids, &val).await?;
                } */
            }

            Task::Sync(ref mut todo) => {
                if shared.is_master() {
                    bail!("current server is master, needn't sync");
                }
                let parent = gfs_hashcode_to_path(&shared.gfs.root, todo.ihash, todo.idx);
                {
                    // Record类型可能存在高频的并发处理相同的文件，可能导致系统崩溃，需要加锁
                    // 归档类不需要，因为不存在修改，只会创建新目录保存新文件。
                    //
                    if todo.code == SYNC_CODE_RECORD {
                        let _record = shared.record.write().await;
                        let writer_down_url = get_download_url(
                            shared.gfs.master.as_str(),
                            todo.idx,
                            todo.ihash,
                            todo.name.as_str(),
                        );
                        if let Err(err) = gfsc_sync_from_master(
                            &writer_down_url,
                            todo.name.as_str(),
                            &parent,
                            todo.name.as_str(),
                        )
                        .await
                        {
                            error!("{:#?}", err);
                        }
                    } else {
                        let writer_down_url = get_download_url(
                            shared.gfs.master.as_str(),
                            todo.idx,
                            todo.ihash,
                            todo.name.as_str(),
                        );
                        if let Ok(()) = gfsc_sync_from_master(
                            &writer_down_url,
                            todo.name.as_str(),
                            &parent,
                            todo.name.as_str(),
                        )
                        .await
                        {
                            shared
                                .gfs
                                .add_stat(todo.ihash, todo.idx, todo.name.as_str())
                                .await?;
                        }
                    }
                }
                extract_and_index(shared, &todo).await?;
            }
            _ => bail!("todo"),
        }
        Ok(())
    }
}

pub(crate) struct Worker {
    pub sync_timer: u64,
    pub shutdown: shutdown::Shutdown,
    /// Not used directly. Instead, when `Handler` is dropped...?
    pub _shutdown_complete: mpsc::Sender<()>,
    pub shared: Arc<ShareComm>,
    pub rmq_client: Arc<ChannelClient>,
}

impl Worker {
    pub async fn run(&mut self) -> Result<()> {
        let mut rx = self.shared.rx.lock().await;
        debug!("start to listen for worker");

        let mut ticker = tokio::time::interval(Duration::from_secs(self.sync_timer));

        let mq_actor: MqActor = if self.shared.mq_server.len() == 0 {
            MqActor::NoneMq()
        } else if self.shared.mq_server.starts_with("redis") {
            let client = redis::Client::open(self.shared.mq_server.as_str())
                .expect("failed to create redis client");
            let mqq = MqRedis {
                client,
                server_name: self.shared.mq_server.clone(),
            };
            MqActor::RedisMq(mqq)
        } else {
            let mqq = MqRabbit {
                server_name: self.shared.mq_server.clone(),
                slave_port: self.shared.port,
                channel_client: self.rmq_client.clone(),
            };
            MqActor::RabbitMq(mqq)
        };

        event!(target:"startup", loglevel(Info), "Worker thread is startup, OK");

        ticker.tick().await; // 首次立刻执行，不合适。
                             //  let shared = DerefMut<SharedHandle>(shared);
        while !self.shutdown.is_shutdown() {
            tokio::select! {
            _ = ticker.tick() => {
                if !self.shared.is_master(){
                   println!("this is slave, then should try to sync");
                   let limit_size:u32 = 10;
                   while !self.shutdown.is_shutdown() {
                        let mut goon = false;
                        let mut sync_id = self.shared.sync_id.lock().await;
                        let mut last = *sync_id;
                        debug!("last id = {}, limit = {}", last, limit_size);
                        let master = gfsc_get_stats(self.shared.gfs.master.as_str(), last, limit_size).await;
                        match master {
                            Ok(master)=>{
                                if master.len() > 0 {
                                    if master.len() == (limit_size as usize) {
                                        goon = true;
                                    }
                                    let local = self.shared.gfs.get_next_from_id(last, limit_size ).await;
                                    match local{
                                        Ok(local) => {
                                            for i in 0..master.len(){
                                                let mut exist = false;
                                                for j in 0..local.len(){
                                                    if master[i].id == local[j].id {
                                                        exist = true;
                                                        break;
                                                    }
                                                }
                                                if !exist{
                                                    error!("{:#?} is not exist, send tx to copy now", master[i] );
                                                    let reps = gfs_stat_as_response(& master[i]);
                                                    if let Err(_) = self.shared.tx.send(Task::Sync(reps)).await{
                                                        error!("receiver dropped, 不可能发生！！");
                                                    }
                                                }
                                            }
                                            last = master[master.len()-1].id;
                                        },
                                        Err(err)=>{
                                            error!("read local stat error:{:#?}", err)
                                        },
                                    }
                                if *sync_id != last {
                                    debug!("update sync_id = {}", last);
                                    *sync_id = last;
                                }
                                }else{
                                    debug!("no data to sync, current is the last: sync_id={}", last);
                                }
                            },
                            Err(err) => { error!("read local stat error:{:#?}", err)  },
                        }
                        if !goon{
                            break;
                        }
                    }
                }
            },
            cmd = rx.recv() => {
                if let Some(mut cmd) = cmd {
                    if let Err(err) = cmd.run( &self.shared, &mq_actor).await {
                        error!("error to dispach task : {:#?}", err);
                    }
                }
            },

            _ = self.shutdown.recv() => {
                    // If a shutdown signal is received, return from `run`.
                    // This will result in the task terminating.
                    //return Ok(());
                break
            }
            }
        }
        event!(target:"shutdown", loglevel(Error), "Worker thread  is shutdown, OK");
        Ok(())
    }
}

/*
pub fn bytes_from_str(src: &str) -> Bytes {
    Bytes::from(src.to_string())
}
*/
/**
对特定文件进行内容抽取，保存为idx.rabbit.txt，对原文件进行sha256 hash,如果已经被index检索，则放弃，否则对内容进行index build。
 - 被抽取的文本保存为idx.rabbit.txt
 - 内容sha256 code保存为docid.sha
 */
pub async fn extract_and_index(shared: &ShareComm, todo: &GfsResponse) -> Result<()> {
    // extract as id.rabbit.txt
    let txt_fname = format!("{}.rabbit.txt", todo.idx);
    let txt_fname = gfs_hashcode_to_path(&shared.gfs.root, todo.ihash, todo.idx).join(txt_fname);
    extract_as_text(
        &shared.gfs.root,
        gfs_hashcode_to_path(&shared.gfs.root, todo.ihash, todo.idx).join(todo.name.as_str()),
        &txt_fname,
    )?;
    // create sha256_code, check index, if exist then return else build it
    let source_file =
        gfs_hashcode_to_path(&shared.gfs.root, todo.ihash, todo.idx).join(todo.name.as_str());
    let sha_code = sha256_file_code(&source_file)?;
    let sha_file = gfs_hashcode_to_path(&shared.gfs.root, todo.ihash, todo.idx).join("docid.sha");
    // save sha256 hashcode as docid.sha
    std::fs::write(sha_file, sha_code.as_bytes())?;

    let index_file = if txt_fname.exists() {
        txt_fname
    } else {
        source_file
    };

    let ext = index_file.extension().unwrap_or_default().to_string_lossy();
    let ext = ext.to_lowercase(); //.as_str();
    if index_file_is_allow(ext.as_str()) {
        let gfs_file = gfs_hashcode_filename(todo.ihash, todo.idx, todo.name.as_str());
        shared
            .index
            .add_file(&index_file, gfs_file.as_str(), sha_code.as_str())
            .await?;
    }
    Ok(())
}

/*
向redis服务器发送消息
 - 这里的发送指令，采用了定长消息，如果redis消息堆积超出定长，将会抛弃堆积的旧消息。

async fn sync_msg_to_mq(client: &redis::Client, topic: &str, key: &str, val: &str) {
    if let Err(err) = redis_message_to_mq(client, topic, key, val).await {
        error!("send sync msg to redis error:  {:#?}", err);
    }
}
*/

/**
对文件内容进行sha256 hash
 - 采用io::copy(&mut file, &mut hasher)?
 - 可比较: let file_byte = std::fs::read(file_name)?; hasher.update(file_byte);
 */
fn sha256_file_code(file_name: &PathBuf) -> Result<String> {
    let mut file = std::fs::File::open(file_name)?;
    let mut hasher = Sha256::new();
    io::copy(&mut file, &mut hasher)?;
    let result = hasher.finalize().to_vec();
    let shacode = hex::encode(result);
    Ok(shacode)
}
/*
#[test]
fn hex_code_test() {
    if let Ok(a) = hex::decode("bd") {
        println!("{:#?}", a[0]);
    }

    let b = hex::encode(vec![189, 1, 255]);
    println!("{:#?}", b);
}

#[test]
fn test_sha256_file_code_no() {
    let source = PathBuf::from("/home/lengss/abc.pptx");
    let s = sha256_file_code(&source).unwrap();
    println!("{}", s);
}

#[test]
fn test_sha256_file_code() {
    let source = PathBuf::from("/home/lengss/hq/abc.pptx");
    let s = sha256_file_code(&source).unwrap();
    println!("{}, len={}", s, s.len());
}
*/
